package hbase.operation;

import hbase.common.MyScan;
import hbase.tools.Constants;
import hbase.tools.HbaseUtils;
import hbase.tools.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class FilterOperation {

    private static Connection conn = null;
    private static Admin admin = null;

    static{
        try {
            Configuration conf = HBaseConfiguration.create();
            conf.set(Constants.ZK_QUORUM,Constants.ZK_QUORUM_NODES);
            conf.set(Constants.ZK_CLIENT_PORT,Constants.ZK_PORT);
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static void scanTableFilterData(String tableName,Filter filter){

        if(HbaseUtils.isTableExists(admin,tableName)){
            Table table = null;
            try {
                table = conn.getTable(TableName.valueOf(tableName));

                MyScan scan = new MyScan();

                //设置查询过滤器
                scan.setFilter(filter);

                //获取需要查询的所有数据的结果对象
                ResultScanner rs = table.getScanner(scan);
                //把结果对象转成迭代器类型，方便后面使用
                Iterator<Result> results = rs.iterator();
                //遍历所有的结果内容
                while(results.hasNext()){
                    Result result = results.next();
                    HbaseUtils.showCellInfo(result);
                }
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                HbaseUtils.closed(conn,table);
            }
        }else{
            System.out.println(tableName + " 表不存在");
        }
    }

    public static void main(String[] args) {

        //行键过滤器 查找 rowKey 值大于 002 的数据
        Filter filter = new RowFilter(CompareOperator.GREATER, new BinaryComparator(Bytes.toBytes("002")));
        //列族过滤器 查找 columnFamily 值等于 baseInfo 的数据
//        Filter filter = new FamilyFilter(CompareOperator.EQUAL,new SubstringComparator("baseInfo"));
        //列过滤器 查找 column 值不以 a 开头的数据
//        Filter filter = new QualifierFilter(CompareOperator.NOT_EQUAL,new BinaryPrefixComparator(Bytes.toBytes("a")));
        //值过滤器 查找 value 值只有两位数字的数据
//        Filter filter = new ValueFilter(CompareOperator.EQUAL,new RegexStringComparator("^\\d{2}$"));

        //时间戳过滤器 查找列表中所有的时间戳的数据
//        List<Long> tsList = new ArrayList<>();
//        tsList.add(1750819752962L);
//        tsList.add(1750819766245L);
//        Filter filter = new TimestampsFilter(tsList);

        //1.单列值过滤器 按照条件，把当前rowkey下的整行内容全部输出
//        SingleColumnValueFilter filter = new SingleColumnValueFilter(
//                Bytes.toBytes("baseInfo"),
//                Bytes.toBytes("name"),
//                CompareOperator.LESS_OR_EQUAL,
//                Bytes.toBytes("smith")
//        );
//        //2.此属性默认为false，表示如果没有找到这个column，此条数据展示；
//        //  此属性可以设置为true，表示如果没有找到这个column，此条数据不展示
//        filter.setFilterIfMissing(true);

        //1.单列值排除器 把查找到此列且值符合条件的这列不展示，此行的其余列展示
//        SingleColumnValueExcludeFilter filter = new SingleColumnValueExcludeFilter(
//                Bytes.toBytes("baseInfo"),
//                Bytes.toBytes("name"),
//                CompareOperator.LESS_OR_EQUAL,
//                Bytes.toBytes("smith")
//        );
//        //2.此属性默认为false，表示如果没有找到这个column，此条数据展示；
//        //  此属性可以设置为true，表示如果没有找到这个column，此条数据不展示
//        filter.setFilterIfMissing(true);

        //行键前缀过滤器
//        Filter filter = new PrefixFilter(Bytes.toBytes("00"));
        //列前缀过滤器
//        Filter filter = new ColumnPrefixFilter(Bytes.toBytes("a"));

        //分页过滤器 获取数据的前n行
        Filter filter1 = new PageFilter(2);


        Filter filterList = new FilterList(filter,filter1);


        FilterOperation.scanTableFilterData("s1",filterList);
    }

}
